package com.ls.fw.index.solr.client.dao.base.impl;

import java.awt.image.BufferedImage;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import javax.imageio.ImageIO;

import net.semanticmetadata.lire.constant.FeatureField;
import net.semanticmetadata.lire.imageanalysis.ColorLayout;
import net.semanticmetadata.lire.imageanalysis.EdgeHistogram;
import net.semanticmetadata.lire.imageanalysis.JCD;
import net.semanticmetadata.lire.imageanalysis.LireFeature;
import net.semanticmetadata.lire.imageanalysis.OpponentHistogram;
import net.semanticmetadata.lire.imageanalysis.PHOG;
import net.semanticmetadata.lire.indexing.hashing.BitSampling;
import net.semanticmetadata.lire.solr.utils.FieldUtils;
import net.semanticmetadata.lire.utils.ImageUtils;
import net.semanticmetadata.lire.utils.SerializationUtils;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

import org.apache.commons.codec.binary.Base64;
import org.apache.http.client.HttpClient;
import org.apache.solr.client.solrj.SolrServer;
import org.apache.solr.client.solrj.impl.CloudSolrServer;
import org.apache.solr.client.solrj.impl.ConcurrentUpdateSolrServer;
import org.apache.solr.client.solrj.impl.HttpClientUtil;
import org.apache.solr.client.solrj.impl.HttpSolrServer;
import org.apache.solr.client.solrj.impl.LBHttpSolrServer;
import org.apache.solr.client.solrj.response.UpdateResponse;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.io.OutputFormat;
import org.dom4j.io.SAXReader;
import org.dom4j.io.XMLWriter;

import com.ls.fw.common.beanparser.bean.PageBean;
import com.ls.fw.common.beanparser.schema.Column;
import com.ls.fw.common.beanparser.schema.Table;
import com.ls.fw.index.bean.Image;
import com.ls.fw.index.solr.client.dao.base.BaseIndexSolrDao;
import com.ls.fw.index.solr.client.exception.IndexSolrException;
import com.ls.fw.index.solr.client.util.HttpUtil;

public class BaseIndexSolrDaoImpl implements BaseIndexSolrDao ,Cloneable{

	public static final String DEFAULT_NAME = "collection1";
	public static final String DEFAULT_COLLECTION = "collection1";
	public static final String HANDLER = "lireq";
	protected static final Map<String,String> schemaPathMap = new HashMap<String,String>(); 
	
	private SolrServer server = null;// solr服务
	
	private String baseURL = "";//http://127.0.0.1/solr/；或者127.0.0.1\:9983
	
	private String name = DEFAULT_NAME;//名称
	
	private boolean autoCommit = false;//是否自动提交
	
	private boolean cloud = false;//是否为分布式
	
	
	public BaseIndexSolrDaoImpl() {
		this(null,DEFAULT_NAME);
	}
	
	public BaseIndexSolrDaoImpl(String baseUrl) {
		this(baseUrl,DEFAULT_NAME);
	}
	
	public BaseIndexSolrDaoImpl(SolrServer server) {
		this.server = server;
	}
	
	public BaseIndexSolrDaoImpl(String baseURL, String name) {
		super();
		this.baseURL = baseURL;
		this.name = name;
		if(baseURL!=null){
			this.server = this.newSolrServer(this.baseURL, this.name);
		}
//		try {
//			BitSampling.readHashFunctions();
//		 } catch (IOException e2) {
//		 	e2.printStackTrace();
//		 } 
	}
	
	@Override
	public UpdateResponse add(Map<String, Object> data) throws IndexSolrException{
		if(data==null){
			throw new IndexSolrException("索引数据不能为空！");
		}
		SolrInputDocument doc = new SolrInputDocument();
		boolean existId = false;
		for(String field : data.keySet()){
			if(field.equals(DEFAULT_PK)){
				existId = true;
			}
			doc.addField(field, data.get(field));
		}
		if(!existId){
			doc.addField(DEFAULT_PK, UUID.randomUUID().toString());
		}
		UpdateResponse res = null;
		try {
			res = this.server.add(doc);
			if(this.autoCommit){
				this.commit();
			}
		} catch (Exception e) {
			if(this.autoCommit){
				this.rollback();
			}
			throw new IndexSolrException("添加索引异常！",e);
		}finally{
		}
		return res;
	}
	
	
	@Override
	public UpdateResponse add(Object model) throws IndexSolrException{
		if(model==null){
			throw new IndexSolrException("索引数据不能为空！");
		}
		Map<String, Object> data = null;
		//TODO:反射对象
		
		return this.add(data);
	}
	
	
	public UpdateResponse save(Object model) throws IndexSolrException{
		return this.add(model);
	}
	
	
	@Override
	public SolrServer buildSolrServer(String baseURL, String name){
		if(this.baseURL.equals(baseURL) && this.name.equals(name)){
			if(this.server!=null){
				return this.server;
			}
		}
		return this.newSolrServer(baseURL,name);
	}
	
	
	@Override
	public SolrServer newSolrServer(String baseUrl, String name){
		if(!baseUrl.startsWith("http:")){
			this.cloud = true;
		}
		if(this.cloud){
			return  this.getCloudSolrServer(baseUrl, name);
		}else{
			return this.getHttpSolrServer(baseUrl);
		}
	}
	
	/**
	 * 该类实例化多用于更新删除索引操作
	 * @param solrServerUrl 是Solr服务器的地址，
	 * @param queueSize 缓冲区大小
	 * @param threadCount 后台线程数
	 * @return
	 * @author ls
	 * 2014年8月19日下午9:34:24
	 */
	protected ConcurrentUpdateSolrServer getConcurrentUpdateSolrServer(String solrServerUrl,  int queueSize,  int threadCount){
		return new ConcurrentUpdateSolrServer(solrServerUrl,queueSize,threadCount);
	}
	
	
	/**
	 * 用于有多个Solr服务器，负载均衡
	 * @param solrServerUrl
	 * @param queueSize
	 * @param threadCount
	 * @return
	 * @author ls
	 * 2014年8月19日下午10:19:19
	 * @throws MalformedURLException 
	 */
	protected LBHttpSolrServer getLBHttpSolrServer(String... solrServerUrls) throws MalformedURLException{
		return new LBHttpSolrServer(solrServerUrls);
	}
	

	/**
	 * 该类实例化多用于查询操作
	 * @param baseUrl
	 * @return
	 * @author ls
	 * 2014年8月19日下午9:35:56
	 */
	protected HttpSolrServer getHttpSolrServer(String baseUrl){
		ModifiableSolrParams params = new ModifiableSolrParams();
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS, 128);
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, 32);
		//params.set(HttpClientUtil.PROP_FOLLOW_REDIRECTS, followRedirects);
		HttpClient httpClient =  HttpClientUtil.createClient(params);
		//该类实例化多用于查询操作
		return new HttpSolrServer(baseUrl+"/"+name,httpClient);

	}

	protected CloudSolrServer getCloudSolrServer(String zkHostString,
			String collection) {
		ModifiableSolrParams params = new ModifiableSolrParams();
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS, 1000);//10
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, 500);//5
		HttpClient client = HttpClientUtil.createClient(params);						
		LBHttpSolrServer lbServer = new LBHttpSolrServer(client);

		CloudSolrServer server1 = null;
		server1 = new CloudSolrServer(zkHostString, lbServer, true);
		server1.setZkClientTimeout(30000);
		server1.setZkConnectTimeout(30000);
		server1.getLbServer().setConnectionTimeout(30000);
		if (collection != null) {
			server1.setDefaultCollection(collection);
		}
		server1.getLbServer().getHttpClient().getParams()
				.setIntParameter("http.connection.timeout", 30000);
		server1.setParallelUpdates(true);
		server1.connect();
		return server1;
	}
	
	@Override
	public SolrServer switchName(String name){
		this.name = name;
		return this.reopen();
	}
	
	
	@Override
	public SolrServer switchServer(String baseUrl,String name){
		this.baseURL = baseUrl;
		return this.switchName(name);
	}
	
	
	@Override
	public SolrServer switchServer(String baseUrl){
		this.baseURL = baseUrl;
		return this.switchName(DEFAULT_NAME);
	}
	
	@Override
	public void commit() throws IndexSolrException {
		try {
			if(server!=null){
				this.server.commit();
			}
		} catch (Exception e) {
			throw new IndexSolrException(e.getMessage(),e);
		}
	}
	
	@Override
	public void rollback() throws IndexSolrException {
		try {
			if(server!=null){
				this.server.rollback();
			}
		} catch (Exception e) {
			throw new IndexSolrException(e.getMessage(),e);
		}
	}
	
	

	@Override
	public SolrServer open()throws IndexSolrException{
		synchronized (this) {
			if(this.server==null){
				this.server = this.newSolrServer(baseURL, name);
			}
		}
		return server;
	}
	
	@Override
	public SolrServer reopen()throws IndexSolrException{
		this.close();
		return this.open();
	}
	
	@Override
	public void close() throws IndexSolrException{
		try{
			this.commit();
		}catch (Exception e) {
			e.printStackTrace();
		}
		try {
			if(server!=null){
				if(this.cloud){
					  ZkStateReader reader = ((CloudSolrServer)server).getZkStateReader();
					  if(reader!=null){
					      reader.close();
					  }
				}
				this.server.shutdown();
			}
			server = null;
		} catch (Exception e) {
			throw new IndexSolrException(e.getMessage(),e);
		}
	}
	
	@Override
	public SolrServer getServer() {
		return server;
	}
	
	public void setServer(SolrServer server) {
		this.server = server;
	}
	
	@Override
	public String getBaseURL() {
		return baseURL;
	}
	
	
	public void setBaseURL(String baseURL) {
		this.baseURL = baseURL;
	}
	
	@Override
	public String getName() {
		return name;
	}
	
	@Override
	public boolean isAutoCommit() {
		return autoCommit;
	}

	private String getSchemaPath(){
		String schemaPath = schemaPathMap.get(this.getName());
		if(schemaPath==null || "".equals(schemaPath)){
			BufferedReader br = null;
			try {
				String schemaPathUrl = this.getBaseURL()+"/admin/cores?wt=json&action=SCHEMAPATH&core="+this.getName()+"&_="+new Date().getTime();
		        URL url = new URL(schemaPathUrl);
		        HttpURLConnection urlConnection = (HttpURLConnection) url.openConnection();
		        InputStream is = urlConnection.getInputStream();  
		        br = new BufferedReader(new InputStreamReader(is));  
		        StringBuilder sb = new StringBuilder();  
		        String str = null;
		        while((str = br.readLine())!=null){  
		            sb.append(str);  
		        }  
		        String content = new String(sb);
		        content = new String(content.getBytes("UTF-8"));
		        JSONObject json = JSONObject.fromObject(content);
		        schemaPath = json.getString("schemePath");
		        JSONObject error = json.getJSONObject("error");
		        json = json.getJSONObject("responseHeader");
		        int status = json.getInt("status");
		        if(status!=0&&error!=null){
		        	throw new IndexSolrException(error.getString("msg"));
		        }else{
		        	schemaPathMap.put(this.getName(), schemaPath);
		        }
			} catch (IndexSolrException e) {
		        throw e;
	        } catch (Exception e) {
		        throw new IndexSolrException("刷新索引异常!",e);
	        } finally{
	        	if(br!=null){
	        		try {
						br.close();
					} catch (IOException e) {
						e.printStackTrace();
					}  
	        	}
	        }
		}
		
		return schemaPath;
	}
	
	@Override
	public void createTable(Table table) throws IndexSolrException {
		SAXReader sax = new SAXReader();
		Document xmlDoc = null;
		String schemaPath = this.getSchemaPath();
		try {
			xmlDoc = sax.read(new File(schemaPath));
		} catch (DocumentException e) {
			throw new IndexSolrException("读取索引schema路径错误",e);
		}
		Element root = xmlDoc.getRootElement();
		Element fieldsElement = root.element("fields");
		if(fieldsElement==null){
			fieldsElement = root;
		}
		List<Column> nodes = table.getColumns();
		for (Column column : nodes) {
			if(fieldsElement.selectNodes("field[@name='"+ column.getIndexName() +"']").size()>0){
				//如果该列已经添加到索引中，则不再添加
				Element reElement = (Element) fieldsElement.selectNodes("field[@name='"+ column.getIndexName() +"']").get(0);
				fieldsElement.remove(reElement);
			}
			Element newField = fieldsElement.addElement("field");
			newField.addAttribute("name", column.getIndexName());
			newField.addAttribute("type", column.getIndexFieldType().getType());
			newField.addAttribute("indexed", String.valueOf(column.isIndex()));
			newField.addAttribute("stored",  String.valueOf(column.isStore()));
		}
		//OutputFormat out = new OutputFormat("  ", true, "utf-8");
		XMLWriter writer = null;
		try {
			OutputFormat format = OutputFormat.createPrettyPrint(); //设置XML文档输出格式
			format.setEncoding("UTF-8");
			format.setNewlines(true);
			writer = new XMLWriter(new FileOutputStream( schemaPath), format);
			writer.write(xmlDoc);
			writer.flush();
			this.reloadIndex();
		} catch (Exception e) {
			throw new IndexSolrException("写回索引schema路径异常",e);
		}finally{
			if(writer!=null){
				try {
					writer.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 刷新索引
	 * @author ls 2013-10-8 下午6:51:18
	 */
	private void reloadIndex() throws IndexSolrException {
		BufferedReader br = null;
		try {
			String reloadUrl = this.getBaseURL()+"/admin/cores?wt=json&action=RELOAD&core="+this.name;
	        URL url = new URL(reloadUrl);
	        HttpURLConnection urlConnection = (HttpURLConnection) url.openConnection();
	        InputStream is = urlConnection.getInputStream();  
            br = new BufferedReader(new InputStreamReader(is));  
            StringBuilder sb = new StringBuilder();  
            String str = null;
	        while((str = br.readLine())!=null){  
	            sb.append(str);  
	        }  
            String content = new String(sb);
            content = new String(content.getBytes("UTF-8"));
            JSONObject json = JSONObject.fromObject(content);
            JSONObject error = json.getJSONObject("error");
	        json = json.getJSONObject("responseHeader");
	        int status = json.getInt("status");
	        if(status!=0 && error!=null){
	        	throw new IndexSolrException(error.getString("msg"));
	        }
        } catch (Exception e) {
	        throw new IndexSolrException("刷新索引异常!",e);
        } finally{
        	if(br!=null){
        		try {
					br.close();
				} catch (IOException e) {
					e.printStackTrace();
				}  
        	}
        }
	}
	@Override
	public void deleteTable(String table) throws IndexSolrException {
		throw new IndexSolrException("暂不支持！");
	}

	@Override
	public Table getTable(String table) throws IndexSolrException {
		throw new IndexSolrException("暂不支持！");
	}

	@Override
	public void updateTable(Table table) throws IndexSolrException {
		throw new IndexSolrException("暂不支持！");
	}

	@Override
	public boolean isCloud() {
		return this.cloud;
	}
	
	
	public BaseIndexSolrDao cloud(boolean cloud){
		this.cloud = cloud;
		return this;
	}
	

	
	public BaseIndexSolrDao name(String name) {
		this.name = name;
		return this;
	}

	public BaseIndexSolrDao autoCommit(boolean autoCommit) {
		this.autoCommit = autoCommit;
		return this;
	}

	@Override
	public void addImage(BufferedImage img, String id,String path, String name)
			throws IndexSolrException {
		this.addImage(img,  id,path, name, null);
	}
	
	private SolrInputDocument getDoc(BufferedImage img, String id,String url, String name) {
		 img = ImageUtils.trimWhiteSpace(img); 
		 SolrInputDocument doc = new SolrInputDocument();
		 doc.addField("id", id);
		 doc.addField("title", name);
		 doc.addField("url", url);
		 FieldUtils.addImageFeature(doc, img);
		return doc;
	}
	
	@Override
	public PageBean queryImage(URL url, int page, int pageSize)
			throws IndexSolrException {
		return this.queryImage(url, page, pageSize, FeatureField.EdgeHistogram);
	}

	@Override
	public PageBean queryImage(URL url, int page, int pageSize, FeatureField field)
			throws IndexSolrException {
		PageBean b = null;
		try {
			StringBuffer sb = new StringBuffer(this.baseURL+"/"+HANDLER);
			sb.append("?");
			sb.append("start="+((page-1)*pageSize));
			sb.append("&rows="+pageSize);
			sb.append("&url="+url);
			sb.append("&field="+field.getHa());
			JSONObject json = HttpUtil.get(sb.toString());
			JSONArray docs = json.getJSONArray("docs");
			List<Image> list = null;
			if(docs!=null){
				list = new ArrayList<Image>(docs.size());
				Iterator<?> iter = docs.iterator();
				Image image = null;
				while(iter.hasNext()){
					image = new Image();
					JSONObject obj = (JSONObject) iter.next();
					if(obj.containsKey("id")){
						image.setId(obj.getString("id"));
					}
					if(obj.containsKey("title")){
						image.setTitle(obj.getString("title"));
					}
					if(obj.containsKey("url")){
						image.setUrl(obj.getString("url"));
					}
					if(obj.containsKey("d")){
						image.setD(obj.getDouble("d"));
					}
					list.add(image);
				}
			}
			int count = json.getInt("RawDocsCount");
//			long rawDocsSearchTime = json.getLong("RawDocsSearchTime");
//			long reRankSearchTime = json.getLong("ReRankSearchTime");
			b = PageBean.getInstance(page, pageSize, count, list);
			json = null;
		} catch (IOException e) {
			throw new IndexSolrException(e.getMessage(),e);
		}
		return b;
	}
	
	public void addImage(URL url,String id,String path,String name) throws IndexSolrException{
		BufferedImage img = null;
		try {
			img = ImageIO.read(url);
		} catch (IOException e) {
			throw new IndexSolrException(e.getMessage(),e);
		}
		this.addImage(img, path, name, null);
	}
	
	public void addImage(BufferedImage img,String id,String path,String name,Map<String,Object> fields) throws IndexSolrException{
		SolrInputDocument doc = this.getDoc(img, id,path, name);
		if(fields!=null && !fields.isEmpty()){
			for(String key:fields.keySet()){
				Object v = fields.get(key);
				if(v!=null){
					doc.addField(key, v);
				}
			}
		}
		 try {
			this.server.add(doc);
		 } catch (Exception e) {
			try {
	            this.server.rollback();
           } catch (Exception e1) {
	            e1.printStackTrace();
           }
			throw new IndexSolrException(e.getMessage(),e);
		 }
		 if(this.autoCommit){
			this.commit();
		 }
	}
	
}
